/*
 * FileName: HBaseOutputFormat.java
 * Author:   zzw
 * Date:     2019年01月25日
 * Description:
 */
package com.chezhibao.flink.output;

import org.apache.flink.api.common.io.OutputFormat;
import org.apache.flink.configuration.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import java.io.IOException;
import java.util.List;

/**
 * 〈〉<br>
 * 〈〉
 *
 * @author zzw
 * @see [相关类/方法]（可选）
 * @since [产品/模块版本]（可选）
 */
public class HBaseOutputFormat implements OutputFormat<Object> {

    private org.apache.hadoop.conf.Configuration conf = null;
    private HTable table = null;
    private String taskNumber = null;
    private int rowNumber = 0;

    private static final long serialVersionUID = 1L;

    @Override
    public void configure(Configuration configuration) {
        this.conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum", "escnode0");
        conf.set("hbase.zookeeper.port", "2181");
    }

    @Override
    public void open(int taskNumber, int numTasks) throws IOException {
        table = new HTable(conf, "tracesql");
        this.taskNumber = String.valueOf(taskNumber);
    }

    @Override
    public void writeRecord(Object object) throws IOException {
        table.put((List<Put>)object);
    }

    @Override
    public void close() throws IOException {
        table.flushCommits();
        table.close();
    }
}
